package com.nami.logic.rocketmq.consumer;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import cn.hutool.json.JSONUtil;
import com.nami.basic.bo.servicetrans.up.UpMsgWrap;
import com.nami.basic.constants.UpTopicConst;
import com.nami.basic.packet.Packet;
import com.nami.basic.packet.message.UpMsgQosReq;
import com.nami.logic.bo.send.PacketTransBO;
import com.nami.logic.qos.QosSendDaemonS2C;
import com.nami.logic.serializer.Serializer;
import com.nami.logic.service.DownMsgSender;
import com.nami.logic.service.PacketFactory;
import com.nami.store.api.service.RemoteL2CacheService;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.dromara.common.core.enums.conver.ConverTypeEnum;
import org.dromara.common.core.utils.IdentityUtil;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@AllArgsConstructor
@RocketMQMessageListener(topic = UpTopicConst.QOS_TOPIC,
    consumerGroup = UpTopicConst.QOS_CONSUMER_GROUP,
    messageModel = MessageModel.BROADCASTING, consumeMode = ConsumeMode.CONCURRENTLY)
public class QosConsumer implements RocketMQListener<String> {

    private final DownMsgSender mqSender;
    private final PacketFactory packetFactory;

    @DubboReference
    private final RemoteL2CacheService cacheService;

    /**
     * 目前为广播
     *
     * @param body
     */
    @Override
    public void onMessage(String body) {
        try {
            TimeInterval timer = DateUtil.timer();
            if (StringUtils.isBlank(body)) {
                log.warn("[qos] message is empty");
                return;
            }
            UpMsgWrap wrap = JSONUtil.toBean(body, UpMsgWrap.class);
            Packet packet = wrap.getPacket();
            UpMsgQosReq request = Serializer.prototype(packet.getSerial()).deserialize(packet.getBody(), UpMsgQosReq.class);

            if (request == null) {
                log.warn("[qos] message is empty");
                return;
            }
            if (CollUtil.isEmpty(request.getMsgIds())) {
                return;
            }

            for (String msgId : request.getMsgIds()) {
                if (StringUtils.isBlank(msgId)) {
                    continue;
                }
                PacketTransBO qosMsg = new PacketTransBO();
                qosMsg.setMessageId(msgId);
                qosMsg.setIdentityNo(IdentityUtil.buildIdentityNo(request.getSenderId(), wrap.getDeviceType()));
                qosMsg.setIfGroupMsg(ConverTypeEnum.C2C.getCode().equals(request.getConverType()) ? false : true);
                QosSendDaemonS2C.getInstance().remove(qosMsg.getFingerprint());
            }
            log.info("[qos] qos处理完毕【userId={}, receiverId={}】size={} use={}ms", request.getSenderId(), request.getReceiverId(),
                request.getMsgIds().size(),
                timer.interval());
        } catch (Exception var2) {
            var2.printStackTrace();
            log.error("[qos] 出现错误：{}", var2);
        }

    }


}
